package org.hyf.inspur.LessonDesin.clear;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.hyf.inspur.LessonDesin.clear.dataclean.TMatcher;
import org.hyf.inspur.LessonDesin.clear.tools.TProperties;

import java.io.IOException;

public class DxFileClear {
    /**
     * @param args[0]：原始数据输入路径 args[1]：结果输出数据
     *                         Description：通过map处理，将不符合规范、特殊格式（配置文件、图片等）数据过滤掉。
     */
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        //设置map内存使用
        Job job = Job.getInstance(conf,"wc");
        job.setJarByClass(DxFileClear.class);
        //map类
        job.setMapperClass(ClearMapper.class);
        //map输出k-v数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        //不使用reduce
        job.setNumReduceTasks(0);
        //结果数据输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setInputFormatClass(TextInputFormat.class);
        //将输入文件切片最小64M，最大128M
        TextInputFormat.setMinInputSplitSize(job, 1024 * 1024 * 128L); // 设置最小分片大小
        TextInputFormat.setMaxInputSplitSize(job, 1024 * 1024 * 128L); // 设置最大分片大小
        //输入路径
        FileInputFormat.addInputPath(job, new Path("d://data/clear"));

        FileOutputFormat.setOutputPath(job, new Path("d://data/output"));
        //提交作业 判断退出条件（0正常退出，1非正常退出）
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class ClearMapper extends Mapper<LongWritable, Text, NullWritable, Text> {

        public void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
            String[] str = value.toString().split(TProperties.getValue("filesplit"));
            // 判断数据字段数量，判断url字段是否有效，去除特定格式
            if ((str.length + "").equals(TProperties.getValue("filelength"))
                    && !"http://".equals(str[14])
                    && !"https://".equals(str[14])
                    && !"".equals(str[14])
                    && !(str[14].toLowerCase()).matches(TProperties.getValue("fileclear"))) {

                StringBuilder sb = new StringBuilder();
                if (!str[14].startsWith("http://") && !str[14].startsWith("https://")) {
                    str[14] = "http://" + str[14];
                }
                // 获取域名
                String domain = str[14].split("/", -1)[2];
                // 去除端口
                if (domain.contains(":")) {
                    domain = domain.split("\\:", -1)[0];
                }
                // 用户号码|一级域名|URL地址
                sb.append(str[1]).append(TProperties.getValue("outfilesplit"))
                        .append(TMatcher.getDomain(domain)).append(TProperties.getValue("outfilesplit"))
                        .append(str[14]);

                //定义输出类型
                Text ovalue = new Text(sb.toString());
                context.write(NullWritable.get(), ovalue);
            }
        }
    }
}